/*
 * Copyright 2021 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.springframework.influxdb;

import java.lang.reflect.Array;
import java.lang.reflect.GenericArrayType;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Pong;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.util.Assert;

public abstract class InfluxDBTemplate<T> extends InfluxDBAccessor implements InfluxDBOperations<T> {

	private Converter<T> converter;
	private final Class<T> measurementClass;

	@SuppressWarnings("unchecked")
	public InfluxDBTemplate(final InfluxDBConnectionFactory connectionFactory) {
		this.measurementClass = (Class<T>) getParameterizedType(this.getClass());
		this.setConverter(new ConverterImpl<T>(measurementClass));
		setConnectionFactory(connectionFactory);
	}

	private static Class<?> getParameterizedType(Class<?> src) {
		Class<?> cls = src;
		List<Type> typeList = new ArrayList<>();
		while (typeList.size() == 0) {
			if (cls.getGenericSuperclass() instanceof ParameterizedType) {
				Type[] types = ((ParameterizedType) cls.getGenericSuperclass()).getActualTypeArguments();
				for (Type type : types) {
					typeList.add(type);
				}
			}
			cls = cls.getSuperclass();
		}
		return getClass(typeList.get(0));
	}

	/**
	 * Get the underlying class for a type, or null if the type is a variable type.
	 *
	 * @param type the type
	 * @return the underlying class
	 */
	@SuppressWarnings("rawtypes")
	private static Class<?> getClass(Type type) {
		if (type instanceof Class) {
			return (Class) type;
		} else if (type instanceof ParameterizedType) {
			return getClass(((ParameterizedType) type).getRawType());
		} else if (type instanceof GenericArrayType) {
			Type componentType = ((GenericArrayType) type).getGenericComponentType();
			Class<?> componentClass = getClass(componentType);
			if (componentClass != null) {
				return Array.newInstance(componentClass, 0).getClass();
			} else {
				return null;
			}
		} else {
			return null;
		}
	}

	public void setConverter(final Converter<T> converter) {
		this.converter = converter;
	}

	@Override
	public void afterPropertiesSet() {
		super.afterPropertiesSet();
		Assert.notNull(converter, "PointCollectionConverter is required");
		this.createDefaultRetentionPolicy();
	}

	/**
	 * 创建默认的保留策略
	 * 
	 * @param 策略名 ：default，保存天数：30天，保存副本数量：1 设为默认保留策略
	 */
	public void createDefaultRetentionPolicy() {
		createRetentionPolicy("default", "30d", 1, true);
	}

	/**
	 * 创建自定义保留策略
	 * 
	 * @param policyName  策略名 "defalut"
	 * @param duration    保存天数 "30d"
	 * @param replication 保存副本数量 1
	 * @param isDefault   是否设为默认保留策略 true
	 */
	public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
		String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
				getDatabase(), duration, replication);
		if (isDefault) {
			sql = sql + " DEFAULT";
		}
		getConnection().query(new Query(sql));
	}

	@SuppressWarnings("deprecation")
	@Override
	public void createDatabase() {
		final String database = getDatabase();
		getConnection().createDatabase(database);
	}

	@Override
	@SuppressWarnings("unchecked")
	public void write(final T... payload) {
		write(Arrays.asList(payload));
	}

	@Override
	public void write(final List<T> payload) {
		final String database = getDatabase();
		//final String retentionPolicy = getConnectionFactory().getProperties().getRetentionPolicy();
		final BatchPoints ops = BatchPoints.database(database)//
				//.retentionPolicy(retentionPolicy)//
				.consistency(InfluxDB.ConsistencyLevel.ALL).build();
		payload.forEach(t -> ops.point(converter.convert(t)));
		getConnection().write(ops);
	}

	public List<T> find(final Query query) {
		QueryResult result = getConnection().query(query);
		return converter.convert(result);
	}

	@Override
	public QueryResult query(final Query query) {
		return getConnection().query(query);
	}

	@Override
	public QueryResult query(final Query query, final TimeUnit timeUnit) {
		return getConnection().query(query, timeUnit);
	}

	@Override
	public void query(Query query, int chunkSize, Consumer<QueryResult> consumer) {
		getConnection().query(query, chunkSize, consumer);
	}

	@Override
	public Pong ping() {
		return getConnection().ping();
	}

	@Override
	public String version() {
		return getConnection().version();
	}
}
